package com.sh.data.engine.common.util;

import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.exception.BusinessException;

import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public abstract class FlinkSqlUtil {

    private static final Map<String, String> MYSQL_TO_FLINK_TYPE_MAP = new HashMap<>();

    // Mapping for PostgreSQL data types to Flink data types
    private static final Map<String, String> POSTGRESQL_TO_FLINK_TYPE_MAP = new HashMap<>();

    // Mapping for Oracle data types to Flink data types
    private static final Map<String, String> ORACLE_TO_FLINK_TYPE_MAP = new HashMap<>();

    private static final Map<String, String> DB2_TO_FLINK_TYPE_MAP = new HashMap<>();

    private static final Map<String, String> SQLSERVER_TO_FLINK_TYPE_MAP = new HashMap<>();


    static {
        // Initialize MySQL to Flink type mappings
        MYSQL_TO_FLINK_TYPE_MAP.put("TINYINT", "TINYINT");
        MYSQL_TO_FLINK_TYPE_MAP.put("SMALLINT", "SMALLINT");
        MYSQL_TO_FLINK_TYPE_MAP.put("MEDIUMINT", "INT");
        MYSQL_TO_FLINK_TYPE_MAP.put("INT", "INT");
        MYSQL_TO_FLINK_TYPE_MAP.put("INTEGER", "INT");
        MYSQL_TO_FLINK_TYPE_MAP.put("BIGINT", "BIGINT");
        MYSQL_TO_FLINK_TYPE_MAP.put("FLOAT", "FLOAT");
        MYSQL_TO_FLINK_TYPE_MAP.put("DOUBLE", "DOUBLE");
        MYSQL_TO_FLINK_TYPE_MAP.put("DECIMAL", "DECIMAL");
        MYSQL_TO_FLINK_TYPE_MAP.put("CHAR", "CHAR");
        MYSQL_TO_FLINK_TYPE_MAP.put("VARCHAR", "VARCHAR");
        MYSQL_TO_FLINK_TYPE_MAP.put("TEXT", "STRING");
        MYSQL_TO_FLINK_TYPE_MAP.put("DATE", "DATE");
        MYSQL_TO_FLINK_TYPE_MAP.put("TIME", "TIME");
        MYSQL_TO_FLINK_TYPE_MAP.put("DATETIME", "TIMESTAMP");
        MYSQL_TO_FLINK_TYPE_MAP.put("TIMESTAMP", "TIMESTAMP");

        // Initialize PostgreSQL to Flink type mappings
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("SMALLINT", "SMALLINT");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("INTEGER", "INT");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("BIGINT", "BIGINT");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("REAL", "FLOAT");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("DOUBLE PRECISION", "DOUBLE");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("NUMERIC", "DECIMAL");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("CHARACTER", "CHAR");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("CHARACTER VARYING", "VARCHAR");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("TEXT", "STRING");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("DATE", "DATE");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("TIME", "TIME");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("TIMESTAMP", "TIMESTAMP");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("TIMESTAMP WITH TIME ZONE", "TIMESTAMP");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("TIMESTAMP WITHOUT TIME ZONE", "TIMESTAMP");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("TIME WITHOUT TIME ZONE", "TIME");
        POSTGRESQL_TO_FLINK_TYPE_MAP.put("BOOLEAN", "BOOLEAN");

        // Initialize Oracle to Flink type mappings
        ORACLE_TO_FLINK_TYPE_MAP.put("NUMBER", "DECIMAL");
        ORACLE_TO_FLINK_TYPE_MAP.put("FLOAT", "FLOAT");
        ORACLE_TO_FLINK_TYPE_MAP.put("BINARY_FLOAT", "FLOAT");
        ORACLE_TO_FLINK_TYPE_MAP.put("BINARY_DOUBLE", "DOUBLE");
        ORACLE_TO_FLINK_TYPE_MAP.put("CHAR", "CHAR");
        ORACLE_TO_FLINK_TYPE_MAP.put("VARCHAR2", "VARCHAR");
        ORACLE_TO_FLINK_TYPE_MAP.put("CLOB", "STRING");
        ORACLE_TO_FLINK_TYPE_MAP.put("DATE", "TIMESTAMP");
        ORACLE_TO_FLINK_TYPE_MAP.put("TIMESTAMP", "TIMESTAMP");
        ORACLE_TO_FLINK_TYPE_MAP.put("TIMESTAMP WITH TIME ZONE", "TIMESTAMP_LTZ");
        ORACLE_TO_FLINK_TYPE_MAP.put("TIMESTAMP WITH LOCAL TIME ZONE", "TIMESTAMP_LTZ");

        DB2_TO_FLINK_TYPE_MAP.put("SMALLINT", "SMALLINT");
        DB2_TO_FLINK_TYPE_MAP.put("INTEGER", "INT");
        DB2_TO_FLINK_TYPE_MAP.put("BIGINT", "BIGINT");
        DB2_TO_FLINK_TYPE_MAP.put("REAL", "FLOAT");
        DB2_TO_FLINK_TYPE_MAP.put("DOUBLE", "DOUBLE");
        DB2_TO_FLINK_TYPE_MAP.put("DECIMAL", "DECIMAL");
        DB2_TO_FLINK_TYPE_MAP.put("CHAR", "CHAR");
        DB2_TO_FLINK_TYPE_MAP.put("VARCHAR", "VARCHAR");
        DB2_TO_FLINK_TYPE_MAP.put("CLOB", "STRING");
        DB2_TO_FLINK_TYPE_MAP.put("DATE", "DATE");
        DB2_TO_FLINK_TYPE_MAP.put("TIME", "TIME");
        DB2_TO_FLINK_TYPE_MAP.put("TIMESTAMP", "TIMESTAMP");
        DB2_TO_FLINK_TYPE_MAP.put("BOOLEAN", "BOOLEAN");


        SQLSERVER_TO_FLINK_TYPE_MAP.put("CHAR", "CHAR");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("VARCHAR", "VARCHAR");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("NVARCHAR", "VARCHAR");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("NCHAR", "VARCHAR");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("TEXT", "STRING");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("NTEXT", "STRING");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("XML", "STRING");

        SQLSERVER_TO_FLINK_TYPE_MAP.put("DECIMAL", "DECIMAL");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("NUMERIC", "NUMERIC");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("MONEY", "DECIMAL");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("SMALLMONEY", "DECIMAL");

        SQLSERVER_TO_FLINK_TYPE_MAP.put("FLOAT", "DOUBLE");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("REAL", "DOUBLE");

        SQLSERVER_TO_FLINK_TYPE_MAP.put("BIT", "BOOLEAN");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("INT", "INT");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("TINYINT", "SMALLINT");  // SQL Server's TINYINT is a 1-byte integer
        SQLSERVER_TO_FLINK_TYPE_MAP.put("SMALLINT", "SMALLINT");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("BIGINT", "BIGINT");

        SQLSERVER_TO_FLINK_TYPE_MAP.put("DATE", "DATE");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("TIME", "TIME");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("DATETIME2", "TIMESTAMP");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("DATETIME", "TIMESTAMP");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("SMALLDATETIME", "TIMESTAMP");
        SQLSERVER_TO_FLINK_TYPE_MAP.put("DATETIMEOFFSET", "TIMESTAMP_LTZ(3)");
    }

    // https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/types/
    public enum FlinkSqlColumnType {
        CHAR,
        VARCHAR,
        STRING,
        BOOLEAN,
        BYTES,
        DECIMAL,
        TINYINT,
        SMALLINT,
        INT,
        BIGINT,
        FLOAT,
        DOUBLE,
        DATE,
        TIME,
        TIMESTAMP,
        TEXT
    }

    private static String replaceType(String sqlType) {
        // 正则表达式，匹配括号及其中的内容
        String regex = "\\(\\d+(,\\d+)?\\)";
        // 替换括号及其中的内容为空字符串
        String result = sqlType.replaceAll(regex, "");
        return result;
    }

    private static String convertTypeWithPrecisionAndScale(String dbType, Map<String, String> typeMap) {
        // Regex to match types with precision and scale, e.g., VARCHAR(50), NUMERIC(10,2)
        Pattern pattern = Pattern.compile("([a-zA-Z]+)\\((\\d+)(,\\d+)?\\)");
        Matcher matcher = pattern.matcher(dbType.toUpperCase());

        if (matcher.matches()) {
            String baseType = matcher.group(1);
            String precision = matcher.group(2);
            String scale = matcher.group(3);

            String flinkType = typeMap.getOrDefault(baseType, "STRING");

            // Append precision and scale if present
            if (scale != null) {
                return flinkType + "(" + precision + scale + ")";
            } else {
                return flinkType + "(" + precision + ")";
            }
        } else {
            // Handle types without precision and scale
            return typeMap.getOrDefault(dbType.toUpperCase(), "STRING");
        }
    }

    public static String convertDbToFlink(String dbType, String sqlType) {
        if (dbType.equalsIgnoreCase(DSType.MySQL.name())) {
            return convertMysqlTypeToFlink(sqlType);
        } else if (dbType.equalsIgnoreCase(DSType.TiDB.name())) {
            return convertMysqlTypeToFlink(sqlType);
        } else if (dbType.equalsIgnoreCase(DSType.PostgreSQL.name())) {
            return convertPostgresqlTypeToFlink(sqlType);
        } else if (dbType.equalsIgnoreCase(DSType.Oracle.name())) {
            return convertOracleTypeToFlink(sqlType);
        } else if (dbType.equalsIgnoreCase(DSType.SQLServer.name())) {
            return convertSqlServerTypeToFlink(sqlType);
        }
        throw new BusinessException("不支持的数据库类型，请联系管理员");
    }

    public static void main(String[] args) {
        String s = convertDbToFlink("postgresql", "NUMERIC(2,4)");
        System.out.println(s);
    }

    public static String convertMysqlTypeToFlink(String mysqlType) {
        return convertTypeWithPrecisionAndScale(mysqlType, MYSQL_TO_FLINK_TYPE_MAP);
    }

    public static String convertPostgresqlTypeToFlink(String postgresqlType) {
        return convertTypeWithPrecisionAndScale(postgresqlType, POSTGRESQL_TO_FLINK_TYPE_MAP);
    }

    public static String convertOracleTypeToFlink(String oracleType) {
        return convertTypeWithPrecisionAndScale(oracleType, ORACLE_TO_FLINK_TYPE_MAP);
    }

    public static String convertSqlServerTypeToFlink(String oracleType) {
        return convertTypeWithPrecisionAndScale(oracleType, SQLSERVER_TO_FLINK_TYPE_MAP);
    }

    public static String dbColumnType2FlinkSqlColumnType(String dbColumnType) {
        // mongo
        if (dbColumnType.toLowerCase().startsWith("boolean")) {
            return FlinkSqlColumnType.BOOLEAN.name();
        }

        // mongo
        if (dbColumnType.toLowerCase().startsWith("string")) {
            return FlinkSqlColumnType.STRING.name();
        }
        if (dbColumnType.toLowerCase().startsWith("clob")) {
            return FlinkSqlColumnType.STRING.name();
        }
        if (dbColumnType.toLowerCase().startsWith("char")) {
            return FlinkSqlColumnType.STRING.name();
        }
        if (dbColumnType.toLowerCase().startsWith("varchar")) {
            return FlinkSqlColumnType.STRING.name();
        }

        if (dbColumnType.toLowerCase().startsWith("smallint")) {
            return FlinkSqlColumnType.INT.name();
        }
        if (dbColumnType.toLowerCase().startsWith("tinyint")) {
            return FlinkSqlColumnType.INT.name();
        }
        // 应对mysql的bit
        if (dbColumnType.toLowerCase().startsWith("bit")) {
            return FlinkSqlColumnType.INT.name();
        }
        // 应对pg的int8
        if (dbColumnType.toLowerCase().startsWith("int")) {
            return FlinkSqlColumnType.INT.name();
        }
        // 应对oracle
        if (dbColumnType.toLowerCase().startsWith("number")) {
            return FlinkSqlColumnType.INT.name();
        }
        // 应对kafka的long
        if (dbColumnType.toLowerCase().startsWith("long")) {
            return FlinkSqlColumnType.INT.name();
        }
        if (dbColumnType.toLowerCase().startsWith("bigint")) {
            return FlinkSqlColumnType.INT.name();
        }
        // 应对oracle
        if (dbColumnType.toLowerCase().startsWith("numeric")) {
            return FlinkSqlColumnType.INT.name();
        }

        // 应对mysql的datetime
        if (dbColumnType.toLowerCase().startsWith("date")) {
            return FlinkSqlColumnType.TIMESTAMP.name();
        }
        if (dbColumnType.toLowerCase().startsWith("timestamp")) {
            return FlinkSqlColumnType.TIMESTAMP.name();
        }
        if (dbColumnType.toLowerCase().startsWith("time")) {
            return FlinkSqlColumnType.TIME.name();
        }

        if (dbColumnType.toLowerCase().startsWith("decimal")) {
            return FlinkSqlColumnType.DOUBLE.name();
        }
        if (dbColumnType.toLowerCase().startsWith("float")) {
            return FlinkSqlColumnType.DOUBLE.name();
        }
        if (dbColumnType.toLowerCase().startsWith("double")) {
            return FlinkSqlColumnType.DOUBLE.name();
        }

        if (dbColumnType.toLowerCase().startsWith("text")) {
            return FlinkSqlColumnType.STRING.name();
        }
        // 要是一个都匹配不上使用原本的
        // return dbColumnType;

        throw new UnsupportedOperationException("表的column type:" + dbColumnType + " 没有对应的flink sql种类");
    }
}
